package com.develop.BusinessGatewayService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;
import com.develop.BeanUtils.ChannelBean;
import com.develop.BeanUtils.KafkaSender;
import com.develop.BeanUtils.Message;
import com.develop.BeanUtils.MsgType;
import com.develop.BeanUtils.SharedStorage;
import com.develop.vo.AlarmMsgVo;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;

/**
 * netty web server handler
 *
 **/

@Component
public class BusinessGatewayServerHandler extends SimpleChannelInboundHandler<AlarmMsgVo> {
	private static final Logger LOG = LoggerFactory.getLogger(BusinessGatewayServerHandler.class);
	/**
	 * 心跳断开次数
	 */
	private int counter = 0;
	private KafkaSender kafkaSender;

	public BusinessGatewayServerHandler(KafkaSender kafkaSender) {
		this.kafkaSender = kafkaSender;
	}

	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
		ctx.flush();
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		ChannelBean channelBean = SharedStorage.getChannelBean(ctx.channel().id());

		LOG.error("异常客户端：CityCode:{},FirmCode:{},OmcType:{}", channelBean.getCityCode(), channelBean.getFirmCode(), channelBean.getOmcType());
		LOG.error("出现异常:", cause);
		ctx.close();
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		super.channelActive(ctx);
	}

	/**
	 * 设备断开
	 */
	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		super.channelInactive(ctx);
		SharedStorage.getChannelList().removeIf(channelBean -> channelBean.getChannelId().equals(ctx.channel().id()));
		LOG.error("-- remove --" + ctx.channel().remoteAddress() + ",共有" + SharedStorage.getChannelList().size() + "个客户端在线");
	}

	/**
	 * 用户状态监听
	 */
	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
		if (evt instanceof IdleStateEvent) {
			IdleStateEvent event = (IdleStateEvent) evt;
			if (event.state().equals(IdleState.READER_IDLE)) {
				if (counter >= 3) {
					// 连续丢失3个心跳包 (断开连接)
					ctx.channel().close().sync();
					SharedStorage.getChannelList().removeIf(channelBean -> channelBean.getChannelId().equals(ctx.channel().id()));
					LOG.error("已与" + ctx.channel().remoteAddress() + "断开连接");
				} else {
					counter++;
					LOG.error(ctx.channel().remoteAddress() + "丢失了第 " + counter + " 个心跳包");
				}
			}
		}
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, AlarmMsgVo alarmMsgVo) throws Exception {
		if (counter > 0) {
			counter = 0;
		}
		LOG.info("业务网关客户端接收到的消息:::" + alarmMsgVo.toString());
		try {
			if (alarmMsgVo.getMsgType() == MsgType.realTimeAlarm.getIndex()) {
				ChannelBean channelBean = SharedStorage.getChannelBean(ctx.channel().id());
				kafkaSender.send(channelBean.getCityCode() + "_" + channelBean.getFirmCode() + "_" + channelBean.getOmcType() + "_realTimeAlarm", alarmMsgVo.getBody());
			} else if (alarmMsgVo.getMsgType() == 1) {
				// 登录业务网关
				JSONObject bodyJSON = JSONObject.parseObject(alarmMsgVo.getBody());

				ChannelBean channelBean = new ChannelBean();
				channelBean.setChannelId(ctx.channel().id());
				channelBean.setChannel(ctx.channel());
				channelBean.setCityCode(bodyJSON.getString("cityCode"));
				channelBean.setFirmCode(bodyJSON.getString("firmCode"));
				channelBean.setOmcType(bodyJSON.getString("omcType"));

				if (SharedStorage.getChannelList().size() == 0 || !SharedStorage.getChannelList().contains(channelBean)) {
					SharedStorage.getChannelList().add(channelBean);

					LOG.error("-- add --" + ctx.channel().remoteAddress() + ",共有" + SharedStorage.getChannelList().size() + "个客户端在线");
				}
			} else if (alarmMsgVo.getMsgType() == MsgType.ackSyncAlarmFileResult.getIndex()) {
				ChannelBean channelBean = SharedStorage.getChannelBean(ctx.channel().id());
				kafkaSender.send(channelBean.getCityCode() + "_" + channelBean.getFirmCode() + "_syncAlarmFile", alarmMsgVo.getBody());
			} else if (alarmMsgVo.getMsgType() == MsgType.reqHeartBeat.getIndex()) {
				/** 响应心跳,保持长连接 */
				StringBuffer msgBodyBuffer = new StringBuffer();
				msgBodyBuffer.append(MsgType.ackHeartBeat + ";");
				msgBodyBuffer.append("reqId=0");

				alarmMsgVo = Message.generateMessage((byte) MsgType.ackHeartBeat.getIndex(), msgBodyBuffer.toString());
				ctx.channel().writeAndFlush(alarmMsgVo);
				LOG.info("向业务网关响应心跳...");
			}
		} finally {
			ReferenceCountUtil.release(alarmMsgVo);
		}
	}
}
